package com.atguigu.app.function;


import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.DWDStaticConstants;
import com.atguigu.utils.DimUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class HbaseSinkDefine extends RichSinkFunction<JSONObject> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(DWDStaticConstants.HBASE_PHOENIX_DRIVER);
        connection = DriverManager.getConnection(DWDStaticConstants.HBASE_PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    // {"sinkTable":""}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // make sql
        String sinkTable = value.getString("sinkTable").toUpperCase();
        JSONObject after = value.getJSONObject("after");
        String sql = generSQL(sinkTable, after);
        if ("update".equals(value.getString("operation"))){
            String key = "DIM:" + sinkTable + ":" + after.getString("id");
            DimUtil.deleteRedisKey(key);
        }
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.executeUpdate();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
    }

    private String generSQL(String sinkTable, JSONObject after) {
        Set<String> keys = after.keySet();
        Collection<Object> values = after.values();
        return "upsert into " + DWDStaticConstants.HBASE_SCHEMA + "." +
                sinkTable + "(" + StringUtils.join(keys, ",") + ")" +
                "values('" + StringUtils.join(values, "','") + "')";
    }
}
